package vlt.el.db.dao;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import vlt.el.db.SQLGenerator;


public class VltDAO {
	
	public void load(Connection connOrig, Connection connDest, String selectOrig, String insertDest, Map<String, String> columns, int rowsBeforeCommit, int thread) throws SQLException{
		PreparedStatement pstmtOrig = connOrig.prepareStatement(selectOrig);
	    ResultSet rsOrig = pstmtOrig.executeQuery();
	    
	    PreparedStatement pstmtDest = connDest.prepareStatement(insertDest);

	    int x=1;
	    try{
	    	while (rsOrig.next()){
		    	if (x < rowsBeforeCommit){
		    		x++;
		    	}else{
		    		System.out.println("         Thread: " + thread + " commit in " + x);
		    		x=1;
		    		connDest.commit();
		    	}
		    	Iterator it = columns.entrySet().iterator();
		    	int parameterIndex = 0;
		    	while (it.hasNext()){
		    		parameterIndex++;
		    		Entry<String, String> column = (Entry<String, String>) it.next();
		    		if ("VARCHAR2".equals(column.getValue())){
		    			pstmtDest.setString(parameterIndex, rsOrig.getString(parameterIndex));
		    		}else if ("NUMBER".equals(column.getValue())){
		    			pstmtDest.setLong(parameterIndex, rsOrig.getLong(parameterIndex));
		    		}else if ("BLOB".equals(column.getValue())){
		    			pstmtDest.setBlob(parameterIndex, rsOrig.getBlob(parameterIndex));
		    		}else if ("CLOB".equals(column.getValue())){
		    			pstmtDest.setClob(parameterIndex, rsOrig.getClob(parameterIndex));
		    		}else if ("DATE".equals(column.getValue())){
		    			pstmtDest.setDate(parameterIndex, rsOrig.getDate(parameterIndex));
		    		}	    		
		    	}
		    	int execute = pstmtDest.executeUpdate();
		    	//System.out.println(execute + " row inserted. line " + x + ", will commit in " + rowsBeforeCommit);
		    }
	    	System.out.println("         Thread: " + thread + " commit in " + x);
    		connDest.commit();
	    } finally {
	    	pstmtOrig.close();
	    	rsOrig.close();
	    	pstmtDest.close();
	    }
	    
	}
	
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void incrementalLoad(Connection connOrig, Connection connDest, String selectOrig, String updateOrigCtl, String insertDest, String deleteDest, List<String> pk, Map<String, String> columns, int rowsBeforeCommit, int thread) throws SQLException{
		System.out.println(selectOrig);
		PreparedStatement pstmtOrig = connOrig.prepareStatement(selectOrig);
	    ResultSet rsOrig = pstmtOrig.executeQuery();
	    
	    System.out.println(insertDest);
	    PreparedStatement pstmtDest = connDest.prepareStatement(insertDest);
	    
		System.out.println(deleteDest);
	    PreparedStatement pstmtDeleteDest = connDest.prepareStatement(deleteDest);
	    
	    System.out.println(updateOrigCtl);
	    PreparedStatement pstmtOrigCtl = connOrig.prepareStatement(updateOrigCtl);

	    int x=1;
	    try{
	    	while (rsOrig.next()){
		    	if (x < rowsBeforeCommit){
		    		x++;
		    	}else{
		    		System.out.println("         Thread: " + thread + " commit in " + x);
		    		x=1;
		    		connDest.commit();
		    	}
		    	
		    	long ctlId = rsOrig.getLong(1);
		    	pstmtOrigCtl.setLong(1, ctlId);
		    	
		    	String typeMod = rsOrig.getString(2);
		    	
		    	Iterator it = columns.entrySet().iterator();
		    	int parameterIndex = 0;
	
		    	while (it.hasNext()){
		    		parameterIndex++;
		    		Entry<String, String> column = (Entry<String, String>) it.next();
		    		if ("VARCHAR2".equals(column.getValue())){
		    			pstmtDest.setString(parameterIndex, rsOrig.getString(parameterIndex+2));
		    			if (("U".equals(typeMod) || "D".equals(typeMod)) && pk.contains(column.getKey())){
		    				pstmtDeleteDest.setString(parameterIndex, rsOrig.getString(parameterIndex+2));
		    			}
		    		}else if ("NUMBER".equals(column.getValue())){
		    			pstmtDest.setLong(parameterIndex, rsOrig.getLong(parameterIndex+2));
		    			if (("U".equals(typeMod) || "D".equals(typeMod)) && pk.contains(column.getKey())){
		    				pstmtDeleteDest.setLong(parameterIndex, rsOrig.getLong(parameterIndex+2));
		    			}
		    		}else if ("BLOB".equals(column.getValue())){
		    			pstmtDest.setBlob(parameterIndex, rsOrig.getBlob(parameterIndex+2));
		    			if (("U".equals(typeMod) || "D".equals(typeMod)) && pk.contains(column.getKey())){
		    				pstmtDeleteDest.setBlob(parameterIndex, rsOrig.getBlob(parameterIndex+2));
		    			}
		    		}else if ("CLOB".equals(column.getValue())){
		    			pstmtDest.setClob(parameterIndex, rsOrig.getClob(parameterIndex+2));
		    			if (("U".equals(typeMod) || "D".equals(typeMod)) && pk.contains(column.getKey())){
		    				pstmtDeleteDest.setClob(parameterIndex, rsOrig.getClob(parameterIndex+2));
		    			}
		    		}else if ("DATE".equals(column.getValue())){
		    			pstmtDest.setDate(parameterIndex, rsOrig.getDate(parameterIndex+2));
		    			if (("U".equals(typeMod) || "D".equals(typeMod)) && pk.contains(column.getKey())){
		    				pstmtDeleteDest.setDate(parameterIndex, rsOrig.getDate(parameterIndex+2));
		    			}
		    		}	    		
		    	}
		    	if ("U".equals(typeMod) || "D".equals(typeMod)){
		    		pstmtDeleteDest.executeUpdate();
		    	}
		    	if ("U".equals(typeMod) || "I".equals(typeMod)){
			    	pstmtDest.executeUpdate();
		    	}
		    	pstmtOrigCtl.executeUpdate();
		    	//System.out.println(execute + " row inserted. line " + x + ", will commit in " + rowsBeforeCommit);
		    }
	    	System.out.println("         Thread: " + thread + " commit in " + x);
    		connDest.commit();
	    } finally {
	    	pstmtOrig.close();
	    	rsOrig.close();
	    	pstmtDest.close();
	    }
	    
	}

}
